package com.lzy.canal.register.consumer;

import com.lzy.canal.component.consumer.MysqlConsumer;
import com.lzy.canal.component.consumer.mysql.DefaultDmlConsumer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * dml insert update消息处理器组件注册器
 *
 */
@Slf4j
public class DmlConsumerRegister implements ConsumerRegister<MysqlConsumer> {

    @Autowired
    private final DefaultDmlConsumer defaultDmlConsumer;

    private Map<String, MysqlConsumer> registerConsumerMap = new ConcurrentHashMap<>();

    public DmlConsumerRegister(DefaultDmlConsumer defaultDmlConsumer) {
        this.defaultDmlConsumer = defaultDmlConsumer;
    }


    @Override
    public ConsumerRegister add(String schemaName, String tableName, MysqlConsumer mysqlConsumer) {
        registerConsumerMap.put(getKey(schemaName, tableName), mysqlConsumer);
        return this;
    }

    @Override
    public MysqlConsumer getConsumer(String schemaName, String tableName) {
        MysqlConsumer consumer = registerConsumerMap.get(getKey(schemaName, tableName));
        if (consumer == null) {
            log.info("未找到消息处理组件,使用默认消息处理组件");
            return defaultDmlConsumer;
        }
        return consumer;
    }
}
